package atguigu.kafka_flume_inteceptor;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;

import java.nio.charset.StandardCharsets;
import java.util.List;

/**
 * 该类用于拦截真实数据产生的时间，
 * 从上游的flume的json数据中获取数据真实的产生时间
 * 生产环境下如果不设置该拦截器，下游的kafka-flume将数据存到hdfs将会使用kafka中topic的时间戳，造成真实数据与落盘数据时间偏差
 */
public class MyTimeInterceptor implements Interceptor {

    @Override
    public void initialize() {

    }

    @Override
    public Event intercept(Event event) {

        // 1. 获取数据字符串
        String bodyJson= new String(event.getBody(), StandardCharsets.UTF_8);
        // 2. 解析为JSON对象
        JSONObject jsonObject = JSON.parseObject(bodyJson);
        // 3. 获取时间戳
        Long ts = jsonObject.getLong("ts");
        // 4. 将时间戳放入头部
        event.getHeaders().put("timestamp",ts.toString());
        return event;
    }
    
    /**
     * 拦截批量事件
     * @param events
     * @return
     */
    @Override
    public List<Event> intercept(List<Event> events) {
        for (Event event : events) {
            //批量拦截事件
            intercept(event);
        }
        return events;
    }

    @Override
    public void close() {

    }

    //构建拦截器的builder，返回拦截器对象给flume进行使用
    public static class Builder implements Interceptor.Builder{

        @Override
        public Interceptor build() {
            return new MyTimeInterceptor();
        }

        @Override
        public void configure(Context context) {

        }
    }

}
